package com.shujia.spark

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator

import scala.collection.mutable.ListBuffer

object Demo18Acc {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("map")


    val sc = new SparkContext(conf)


    val student: RDD[String] = sc.textFile("spark/data/students.txt")


    /**
      * 算子内的代码运行在Executor，算子外面的代码运行在Driver端
      * 在算子内修改算子外的变量不会生效
      *
      */
    var i = 0
    student.foreach(line => {
      i += 1
      println(line)
    })
    println(i)


    /**
      * 累加器
      * 1、在每一个task中进行局部累加
      * 2、当job执行完成之后，在driver端进行汇总
      */

    //定义一个累加器
    val accumulator: LongAccumulator = sc.longAccumulator


    student.foreach(line => {
      //累加器可以在算子内进行累加
      accumulator.add(1)
      println(line)
    })

    //读取累加器的结果

    println(accumulator.value)


  /*  student.foreach(line => {

      //在spark的算子内不能再使用rdd
      //因为算子中的代码会被封装成一个task发生到Executor中执行，rdd 本身就是一个弹性的分布式数据集，不能在网络中传输
      student.foreach(l => {
        println(l)
      })

    })*/

  }

}
